package messageOrderingModule;

import generalObjects.GroupView;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.ReentrantLock;

import communicationModule.MessageObject;
import communicationModule.SendResult;

/**
 * Constitutes the GCom implementation of a causal ordering algorithm. This
 * class sorts messages according to the causal algorithm(happened-before).
 * Messages are sent to an object of this class. They are then sorted in an
 * internal structure according to the causal algorithm. Any messages that are
 * considered to be in order are returned for delivery.
 * 
 */
public class CausalAlgorithm implements OrderingAlgorithm {

    private final List<SendResult> deliverHoldbackQueue;
    private final GroupView gView;
    private final ReentrantLock lock;
    private final AtomicBoolean stateChanged;

    public CausalAlgorithm(final GroupView gView, AtomicBoolean stateChanged) {
        this.gView = gView;
        this.stateChanged = stateChanged;
        lock = new ReentrantLock();
        deliverHoldbackQueue = new ArrayList<SendResult>();
    }

    @Override
    public List<SendResult> getDeliverables(final List<SendResult> resultList) {

        boolean addToList = false;
        boolean insequence = false;
        boolean startedOver = false;
        int added = 0;
        final ConcurrentHashMap<String, Integer> myVector = gView
                .getVectorClock();
        final HashMap<Integer, String> check = new HashMap<Integer, String>();
        final List<SendResult> delivery = new ArrayList<SendResult>();
        final List<SendResult> removableItems = new ArrayList<SendResult>();

        ArrayList<String> incrementKeys = new ArrayList<String>();
        if (resultList != null) {
            for (final SendResult result : resultList) {
                lock.lock();
                deliverHoldbackQueue.add(result);
                lock.unlock();
                stateChanged.compareAndSet(false, true);
            }
        }

        for (int i = 0; i < deliverHoldbackQueue.size(); i++) {

            if (startedOver) {
                added = 0;
                startedOver = false;
                for (int k = 0; k < incrementKeys.size(); k++) {
                    gView.incrementSeq(incrementKeys.get(k));
                }
                incrementKeys = new ArrayList<String>();
            }
            if (!check.containsKey(i)) {
                lock.lock();
                final SendResult result = deliverHoldbackQueue.get(i);
                lock.unlock();
                final String sender = result.getMsg().getSender();
                if (!result.getMsg().getHeader().equals("message")) {
                    delivery.add(result);
                    check.put(i, null);
                    continue;
                }
                addToList = true;
                insequence = false;
                final ConcurrentHashMap<String, Integer> other = result
                        .getMsg().getGroupView().getVectorClock();

                for (final Map.Entry<String, Integer> entry : myVector
                        .entrySet()) {
                    int otherSN;
                    if (other.get(entry.getKey()) != null) {
                        otherSN = other.get(entry.getKey());
                    } else {
                        otherSN = 0;
                    }
                    final int mySN = entry.getValue();
                    // System.out.println(mySN + " " + otherSN);

                    if (entry.getKey().equals(sender) && (otherSN <= mySN)) {
                        removableItems.add(result);
                        continue;
                    }
                    // if senders sequence number is next in sequence
                    if (entry.getKey().equals(sender)
                            && (otherSN == (mySN + 1))) {
                        insequence = true;
                        continue;
                    }
                    if (otherSN > mySN) {

                        addToList = false;
                        break;
                    }
                }

                if (addToList && insequence) {
                    incrementKeys.add(sender);
                    delivery.add(result);

                    check.put(i, null);
                    added++;
                }
            }
            if (i == deliverHoldbackQueue.size() - 1 && added > 0) {
                i = -1;
                startedOver = true;
            }
        }
        lock.lock();
        for (final SendResult s : delivery) {
            deliverHoldbackQueue.remove(s);
        }
        for (final SendResult s : removableItems) {
            deliverHoldbackQueue.remove(s);
        }
        lock.unlock();
        if (!removableItems.isEmpty() || !delivery.isEmpty()) {
            stateChanged.compareAndSet(false, true);
        }
        return delivery;
    }

    public MessageObject[] getHoldbackQueue() {
        MessageObject[] holdback;
        lock.lock();
        holdback = new MessageObject[deliverHoldbackQueue.size()];
        for (int i = 0; i < deliverHoldbackQueue.size(); i++) {
            holdback[i] = deliverHoldbackQueue.get(i).getMsg();
        }
        lock.unlock();
        return holdback;
    }
}
